package cas.ihep.alluxio.impl;

import alluxio.AlluxioURI;
import alluxio.Configuration;
import alluxio.PropertyKey;
import alluxio.client.block.BlockMasterClient;
import alluxio.client.file.FileSystemMasterClient;
import alluxio.client.file.options.CompleteFileOptions;
import alluxio.client.file.options.CreateFileOptions;
import alluxio.client.file.options.GetStatusOptions;
import alluxio.client.file.options.OutStreamOptions;
import alluxio.client.file.policy.FileWriteLocationPolicy;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.PreconditionMessage;
import alluxio.exception.status.UnavailableException;
import alluxio.resource.CloseableResource;
import alluxio.util.CommonUtils;
import alluxio.util.network.NettyUtils;
import alluxio.wire.LoadMetadataType;
import alluxio.wire.WorkerNetAddress;
import cas.ihep.alluxio.AlluxioBlock;
import cas.ihep.alluxio.AlluxioFile;
import com.google.common.base.Preconditions;

import java.io.IOException;

public class AlluxioFileOut extends AlluxioBaseFile {

    private OutStreamOptions mCreateOptions;

    AlluxioFileOut(AlluxioBaseSystem sys, AlluxioURI path) {
        super(sys, path);
    }

    @Override
    public AlluxioFile open() throws IOException {
        //FileSystemMasterClient masterClient=mSys.mCtx.acquireMasterClient();
        CreateFileOptions options=CreateFileOptions.defaults();
        try (CloseableResource<FileSystemMasterClient> masterClient = mSys.getContext()
                .acquireMasterClientResource()) {
            masterClient.get().createFile(mUri, options);
            mStatus=masterClient.get().getStatus(mUri, GetStatusOptions.defaults().setLoadMetadataType(LoadMetadataType.Never));
        }
        mCreateOptions=options.toOutStreamOptions();
        mCreateOptions.setUfsPath(mStatus.getUfsPath());
        mCreateOptions.setMountId(mStatus.getMountId());
        mBlockSize=options.getBlockSizeBytes();
        return this;
    }

    @Override
    public AlluxioBlock getBlock(int idx) {
        long id=mStatus.getBlockIds().get(idx);
        if (mBlocks.containsKey(id)){
            return mBlocks.get(id);
        }
        return null;
    }

    public void close()throws IOException{
        if (mClosed){
            return;
        }
        mClosed=true;
        try(CloseableResource<FileSystemMasterClient> masterClient = mSys.getContext().acquireMasterClientResource()){
            CompleteFileOptions options=CompleteFileOptions.defaults();
            masterClient.get().completeFile(this.mUri, options);
        }finally {
            mCloser.close();
        }
    }
    @Override
    public boolean canRead() {
        return false;
    }

    @Override
    public boolean canWrite() {
        return true;
    }

    public AlluxioWritableBlock nextBlock()throws IOException{
        long nextid=getNextBlockId();
        WorkerNetAddress address;
        FileWriteLocationPolicy locationPolicy = Preconditions.checkNotNull(mCreateOptions.getLocationPolicy(),
                PreconditionMessage.FILE_WRITE_LOCATION_POLICY_UNSPECIFIED);
        address = locationPolicy.getWorkerForNextBlock(mSys.getInternalHandler().getWorkerInfoList(), mBlockSize);
        if (address == null) {
            throw new UnavailableException(
                    ExceptionMessage.NO_SPACE_FOR_BLOCK_ON_WORKER.getMessage(mBlockSize));
        }
        if (mBlockSize == -1) {
            try (CloseableResource<BlockMasterClient> blockMasterClientResource =
                         mSys.getContext().acquireBlockMasterClientResource()) {
                mBlockSize = blockMasterClientResource.get().getBlockInfo(nextid).getLength();
            }
        }
        if (CommonUtils.isLocalHost(address) && Configuration
                .getBoolean(PropertyKey.USER_SHORT_CIRCUIT_ENABLED) && !NettyUtils
                .isDomainSocketSupported(address)) {
            //LOG.debug("Creating short circuit output stream for block {} @ {}", getBlockId, address);
            //return LocalFilePacketWriter.create(mSys.mCtx, address, nextid, options);
            AlluxioWritableBlock blk=new AlluxioWritableBlock(this,address,nextid,mBlockSize,mCreateOptions);
            mBlocks.put(nextid,blk);
            return mCloser.register(blk);
        }
        throw new IOException("Unable to allocate local block");
    }

    private long getNextBlockId() throws IOException {
        try (CloseableResource<FileSystemMasterClient> masterClient = mSys.getContext()
                .acquireMasterClientResource()) {
            return masterClient.get().getNewBlockIdForFile(mUri);
        }
    }
}
